/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.transport.udp;

import java.util.Enumeration;
import java.util.Hashtable;

import cgl.narada.util.ByteUtilities;

/**
 * <p>
 * Title: UDPPingHandler
 * </p>
 * <p>
 * Description: This class sends a ping message to figure out a node is up or
 * down.
 * </p>
 * 
 * @author Beytullah Yildiz
 * @version 1.0
 */

public class UDPPingHandler extends Thread {

    private Hashtable linkshashtable;

    private Hashtable pinghashtable;

    private UDPLinkFactory udpLinkFactory;

    private String moduleName = "UDPPingHandler: ";

    private boolean udpPingDebug = false;

    private int maxpingfailnumber = 5;

    private long pinginterval = 1000;

    private boolean keepLooping = true;

    public UDPPingHandler() {
        setName("UDPPingHandler");
    }

    // public UDPPingHandler(Hashtable links, UDPLinkFactory udpLinkFactory) {
    // this.linkshashtable = links;
    // pinghashtable = new Hashtable();
    // this.udpLinkFactory = udpLinkFactory;
    //
    // }

    public UDPPingHandler(UDPLinkFactory udpLinkFactory) {
        this.linkshashtable = new Hashtable();
        pinghashtable = new Hashtable();
        this.udpLinkFactory = udpLinkFactory;
        setName("UDPPingHandler");
    }

    public void shutdownServices() {
        keepLooping = false;
        interrupt();
    }

    public void addLinkForMonitoring(UDPLink link) {
        if (udpPingDebug)
            System.out.println(moduleName + "==> Adding Link");

        this.linkshashtable.put(link.getLinkId(), link);

    }

    /**
     * sets the veriable to value to limit the allowable number of sequential
     * ping fail
     * 
     * @param number
     *            maximum allowable number
     */
    public void setMaxPingFailNumber(int number) {
        maxpingfailnumber = number;
    }

    /**
     * @return the maximum number of failed ping
     */
    public int getMaxPingFailNumber() {
        return maxpingfailnumber;
    }

    /**
     * sets a debug value for this class
     * 
     * @param debug
     */
    public void setDebug(boolean debug) {
        udpPingDebug = debug;
    }

    /**
     * sets the value of interval between two pings.
     * 
     * @param time
     *            is an interval values
     */
    public void setPingInterval(long time) {
        pinginterval = time;
    }

    /**
     * @return the time for between two ping messages
     */
    public long getPingInterval() {
        return pinginterval;
    }

    /**
     * calculetes the RTT value between two nodes. calculation is done by adding
     * half of previous rtt and half of current rtt
     * 
     * @param pingobject
     *            keeps whole data about a ping
     * @param rtt
     *            RTT value for between two nodes.
     */
    public void calculateNewRTT(PingObject pingobject, long rtt) {
        long longvalue = pingobject.getValueofRTT();
        if (longvalue != 0) {
            longvalue = longvalue / 2;
            longvalue = longvalue + (rtt / 2);
        } else {
            longvalue = rtt;
        }
        pingobject.setValueofRTT(longvalue);
    }

    /**
     * creates ping message. it consists of header for ping message and sending
     * time.
     * 
     * @param pmessage
     *            byte array of ping message
     * @param time
     *            sending time. it is used both calculation for rtt and
     *            seperation between ping messages
     */
    public void pingMessage(byte[] pmessage, byte[] time) {

        pmessage[0] = (byte) 104;
        System.arraycopy(time, 0, pmessage, 1, time.length);
    }

    /**
     * removes a ping object from hashtable so that the ping for that not is not
     * required any more.
     * 
     * @param linkname
     *            the name of link for a node
     */
    public void removePing(String linkname) {
        pinghashtable.remove(linkname);
        if (udpPingDebug) {
            System.out.println(moduleName + "Link <" + linkname + "> is down");
            System.out.println("The number of links is decresed to "
                + pinghashtable.size());
        }

    }

    /**
     * does the required tasks when an acknowledge received for a ping message
     * 
     * @param linkname
     *            the name of link whose acknowledge is received
     * @param pingtime
     *            the time of ping whose acknowledge is received
     */
    public void ackReceived(String linkname, long pingtime) {
        if (udpPingDebug) {
            System.out.println(moduleName + " ping ack is received for link "
                + linkname + " with received time " + pingtime);
        }

        PingObject po = (PingObject) pinghashtable.get(linkname);
        if (po != null) {
            long currenttime = System.currentTimeMillis();
            po.setLastAckTime(currenttime);
            po.setAcklessPingNumber(0);
            // long rtttime = currenttime - pingtime;
            // calculateNewRTT(po, rtttime);
            // if(udpPingDebug){
            // System.out.println("The RTT of link <" +po.getLinkName()+"> "+
            // po.getValueofRTT());
            // }
        }

    }

    /**
     * sends a ping message to a given node
     * 
     * @param link
     *            consists information a node to send a ping
     * @param _pingtime
     *            ping time for the ping
     * @throws Exception
     */
    public void sendPing(UDPLink link, long _pingtime) throws Exception {
        byte[] pingmessage = new byte[9];
        byte[] bytepingtime = ByteUtilities.getBytes(_pingtime);
        pingMessage(pingmessage, bytepingtime);
        link.sendData(pingmessage);

        if (udpPingDebug) {
            System.out.println(moduleName + " " + pingmessage[0]
                + " ping message is sent for link " + link.getLinkId()
                + " with time " + _pingtime);
        }

    }

    /**
     * sends ping in a given interval if acknowledge to a ping received, does
     * required task to handle acknowledge
     * 
     * @throws Exception
     */
    public void handlePing() throws Exception {

        Enumeration e = linkshashtable.elements();

        while (e.hasMoreElements()) {

            UDPLink link = (UDPLink) e.nextElement();
            String linkname = link.getLinkId();

            // create new object for ping
            if (pinghashtable.get(linkname) == null) {
                PingObject po = new PingObject();
                long currenttime = System.currentTimeMillis();

                po.setLinkName(linkname);
                po.setLastAckTime(0);
                po.setPingTime(currenttime);
                po.setAcklessPingNumber(0);
                pinghashtable.put(linkname, po);

                sendPing(link, currenttime);
                if (udpPingDebug) {
                    System.out.println("The number of links is increased to "
                        + pinghashtable.size());
                }

                continue;
            }

            if (pinghashtable.get(linkname) != null) {
                PingObject po = (PingObject) pinghashtable.get(linkname);

                if (po.getAcklessPingNumber() >= maxpingfailnumber) {

                    UDPLink udplink = (UDPLink) linkshashtable.get(linkname);
                    removePing(udplink.getLinkId());
                    udpLinkFactory.reportLinkLoss(udplink);

                } else {

                    long currenttime = System.currentTimeMillis();
                    po.setPingTime(currenttime);
                    sendPing(link, currenttime);
                    po.incrementAcklessPing();
                    if (udpPingDebug) {
                        System.out
                            .println("The number of ackless ping after sending a ping "
                                + po.getAcklessPingNumber());
                    }

                }
            }

        }
    }

    /**
     * runs thread
     */
    public void run() {
        while (keepLooping) {
            try {
                handlePing();
                try {
                    sleep(pinginterval);
                } catch (InterruptedException ie) {
                    if (keepLooping) {
                        System.err
                            .println(moduleName + "Problem sleeping" + ie);
                    }
                }
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }

        if (udpPingDebug) {
            System.out.println(moduleName + "Shutting down ...");
        }
    }

    /**
     * 
     * <p>
     * PingObject:
     * </p>
     * <p>
     * Description: This class is to keep information about a ping.
     * </p>
     * 
     * @author Beytullah Yildiz
     * @version 1.0
     */
    class PingObject {
        long pingtime;

        long lastacktime;

        int acklesspingnumber;

        long valueofRTT;

        String linkname;

        public PingObject() {
        }

        public void setPingTime(long time) {
            pingtime = time;
        }

        public long getPingTime() {
            return pingtime;
        }

        public void setLastAckTime(long time) {
            lastacktime = time;
        }

        public long getLastAckTime() {
            return lastacktime;
        }

        public void setAcklessPingNumber(int number) {
            acklesspingnumber = number;
        }

        public void incrementAcklessPing() {
            acklesspingnumber++;
        }

        public int getAcklessPingNumber() {
            return acklesspingnumber;
        }

        public void setValueofRTT(long time) {
            valueofRTT = time;
        }

        public long getValueofRTT() {
            return valueofRTT;
        }

        public void setLinkName(String name) {
            linkname = name;
        }

        public String getLinkName() {
            return linkname;
        }

        public void setParameters(long time, int number, String name) {
            pingtime = time;
            acklesspingnumber = number;
            linkname = name;
        }

    }

    public static void main(String[] args) {
    }

}
